package com.sdhs.paas.flux.app;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;

import com.sdhs.paas.flux.model.MyEventListener;
import com.sdhs.paas.flux.subscribe.SampleSubscriber;

import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
import reactor.core.scheduler.Schedulers;

public class TestFluxDemo {
	public static void main(String[] args) throws InterruptedException {
		Flux<String> seq1 = Flux.just("foo", "bar", "foobar");

		List<String> iterable = Arrays.asList("foo", "bar", "foobar");
		Flux<String> seq2 = Flux.fromIterable(iterable);
		
		Mono<String> noData = Mono.empty(); 

		Mono<String> data = Mono.just("foo");

		Flux<Integer> numbersFromFiveToSeven = Flux.range(5, 3); 
		Flux<Integer> ints = Flux.range(1, 3); 
		ints.subscribe(); 
		ints.subscribe(i -> System.out.println(i));
		System.out.println("------------------Error handler--------------------------");
//		Flux<Integer> intss = Flux.range(1, 4) 
//			      .map(i -> { 
//			        if (i <= 3) return i; 
//			        throw new RuntimeException("Got to 4"); 
//			      });
//		intss.subscribe(i -> System.out.println(i), 
//			      error -> System.err.println("Error: " + error));
		System.out.println("------------------complete handler--------------------------");
		Flux<Integer> intsss = Flux.range(1, 4); 
		intsss.subscribe(i -> System.out.println(i),
		    error -> System.err.println("Error " + error),
		    () -> {System.out.println("Done");}); 
		System.out.println("------------------custom subscribe--------------------------");
		SampleSubscriber<Integer> ss = new SampleSubscriber<Integer>();
		Flux<Integer> intssss = Flux.range(1, 4);
//		intssss.subscribe(i -> System.out.println(i),
//		    error -> System.err.println("Error " + error),
//		    () -> {System.out.println("Done");},
//		    s -> ss.request(10));
		intssss.subscribe(ss);
		System.out.println("-----------------flux generate------------------------");
		Flux<String> flux = Flux.generate(
			    () -> 0, 
			    (state, sink) -> {
			      sink.next("3 x " + state + " = " + 3*state); 
			      if (state == 10) {sink.complete(); return state;} 
			      return state + 1; 
			    });
		flux.subscribe(i -> System.out.println(i));
		System.out.println("-------------------flux atomiclong-------------------------------");
		Flux<String> flux1 = Flux.generate(
			    AtomicLong::new, 
			    (state, sink) -> {
			      long i = state.getAndIncrement(); 
			      sink.next("3 x " + i + " = " + 3*i);
			      if (i == 10) sink.complete();
			      return state; 
			    });
		flux1.subscribe(i -> System.out.println(i),
				error -> System.err.println("error"+error),
				() -> System.out.println("done"));
		
		
		System.out.println("-------------------flux subscribe customer----------------------------");
		Flux<String> flux2 = Flux.generate(
			    AtomicLong::new,
			      (state, sink) -> { 
			      long i = state.getAndIncrement(); 
			      sink.next("3 x " + i + " = " + 3*i);
			      if (i == 10) sink.complete();
			      return state; 
			    }, (state) -> System.out.println("state: " + state)); 
		flux2.subscribe(i -> System.out.println(i));
		System.out.println("-------------flux create demo--------------------------------");
//		Flux<String> bridge = Flux.create(sink -> {
//		    myEventProcessor.register( 
//		      new MyEventListener<String>() { 
//
//		        public void onDataChunk(List<String> chunk) {
//		          for(String s : chunk) {
//		            sink.next(s); 
//		          }
//		        }
//
//		        public void processComplete() {
//		            sink.complete(); 
//		        }
//		    });
//		});
//		Flux<String> bridge = Flux.create(sink -> {
//		    sink.onRequest(n -> channel.poll(n))
//		        .onCancel(() -> channel.cancel()) 
//		        .onDispose(() -> channel.close());  
//		    });
		
		Flux<String> alphabet = Flux.just(-1, 30, 13, 9, 20)
			    .handle((i, sink) -> {
			        String letter = alphabet(i); 
			        if (letter != null) 
			            sink.next(letter); 
			    });

			alphabet.subscribe(System.out::println);
		System.out.println("-------------publish on----------------------------");
		Flux<Integer> result = Flux.range(1, 10000) 
	    .publishOn(Schedulers.parallel()) ;
	    result.subscribe(i -> System.out.println(i));
	    System.out.println("--------------handler error-----------------------");
	    AtomicBoolean isDisposed = new AtomicBoolean();
	    Disposable disposableInstance = new Disposable() {
	        @Override
	        public void dispose() {
	            isDisposed.set(true); 
	        }

	        @Override
	        public String toString() {
	            return "DISPOSABLE";
	        }
	    };
	    Flux<String> fluxdis =
	    		Flux.using(
	    		        () -> disposableInstance, 
	    		        disposable -> Flux.just(disposable.toString()), 
	    		        Disposable::dispose 
	    		);
	    fluxdis.subscribe(i -> System.out.println(i));
	    System.out.println("-------------finally---------------------------");
	    LongAdder statsCancel = new LongAdder(); 
	    Flux<String> fluxlong =
	    		Flux.just("foo", "bar")
	    		    .doFinally(type -> {
	    		        if (type == SignalType.CANCEL) 
	    		          statsCancel.increment(); 
	    		    })
	    		    .take(1);
	    System.out.println("----------sleep ---------------");
	    Flux<String> fluxsleep =
	    		Flux.interval(Duration.ofMillis(250))
	    		    .map(input -> {
	    		        if (input < 3) return "tick " + input;
	    		        throw new RuntimeException("boom");
	    		    })
	    		    .onErrorReturn("Uh oh");

	    		fluxsleep.subscribe(i -> System.out.println(i));
	    		Thread.sleep(2100);
	}
	
	public static String alphabet(int letterNumber) {
        if (letterNumber < 1 || letterNumber > 26) {
                return null;
        }
        int letterIndexAscii = 'A' + letterNumber - 1;
        return "" + (char) letterIndexAscii;
	}
}
